/*-------------------------------------------------------------------------
 *
 * createas.c
 *	  执行 CREATE TABLE ... AS，亦称为 SELECT INTO。
 *	  由于 CREATE MATERIALIZED VIEW 共享语法和大多数行为，
 *	  我们也在这里实现它。
 *
 * 我们通过将查询的正常输出转发到专门的 DestReceiver 类型来实现。
 *
 * 以前，CTAS 被作为 SELECT 的变体来实现，这导致各种遗留行为，我们仍然试图保留，特别是我们必须在 QueryCompletion 中返回处理的元组计数。  （不过，我们不再在 CTAS ... WITH NO DATA 中这样做。）
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/commands/createas.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/reloptions.h"
#include "access/sysattr.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "catalog/namespace.h"
#include "catalog/toasting.h"
#include "commands/createas.h"
#include "commands/matview.h"
#include "commands/prepare.h"
#include "commands/tablecmds.h"
#include "commands/view.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
#include "parser/parse_clause.h"
#include "rewrite/rewriteHandler.h"
#include "storage/smgr.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/rls.h"
#include "utils/snapmgr.h"

typedef struct
{
	DestReceiver pub;			/* 公共已知的函数指针 */
	IntoClause *into;			/* 目标关系规范 */
	/* 这些字段由 intorel_startup 填充： */
	Relation	rel;			/* 写入的关系 */
	ObjectAddress reladdr;		/* ExecCreateTableAs 的 rel 地址 */
	CommandId	output_cid;		/* 插入输出元组的 cmin */
	int			ti_options;		/* table_tuple_insert 性能选项 */
	BulkInsertState bistate;	/* 批量插入状态 */
} DR_intorel;

/* CTAS 定义创建的实用函数 */
static ObjectAddress fc_create_ctas_internal(List *fc_attrList, IntoClause *fc_into);
static ObjectAddress fc_create_ctas_nodata(List *fc_tlist, IntoClause *fc_into);

/* 用于收集数据的 DestReceiver 例程 */
static void fc_intorel_startup(DestReceiver *fc_self, int fc_operation, TupleDesc fc_typeinfo);
static bool fc_intorel_receive(TupleTableSlot *fc_slot, DestReceiver *fc_self);
static void fc_intorel_shutdown(DestReceiver *fc_self);
static void fc_intorel_destroy(DestReceiver *fc_self);


/*
 * create_ctas_internal
 *
 * 用于通过 CREATE TABLE AS 或物化视图创建关系定义的内部工具。
 * 调用者需要提供一个属性列表（ColumnDef 节点）。
 */
static ObjectAddress fc_create_ctas_internal(List *fc_attrList, IntoClause *fc_into)
{
	CreateStmt *fc_create = makeNode(CreateStmt);
	bool		fc_is_matview;
	char		fc_relkind;
	Datum		fc_toast_options;
	static char *fc_validnsps[] = HEAP_RELOPT_NAMESPACES;
	ObjectAddress fc_intoRelationAddr;

	/* 该代码支持 CREATE TABLE AS 和 CREATE MATERIALIZED VIEW */
	fc_is_matview = (fc_into->viewQuery != NULL);
	fc_relkind = fc_is_matview ? RELKIND_MATVIEW : RELKIND_RELATION;

	/*
	 * 通过伪造 CREATE TABLE 解析树来创建目标关系，并将其
	 * 传递给 DefineRelation。
	 */
	fc_create->relation = fc_into->rel;
	fc_create->tableElts = fc_attrList;
	fc_create->inhRelations = NIL;
	fc_create->ofTypename = NULL;
	fc_create->constraints = NIL;
	fc_create->options = fc_into->options;
	fc_create->oncommit = fc_into->onCommit;
	fc_create->tablespacename = fc_into->tableSpaceName;
	fc_create->if_not_exists = false;
	fc_create->accessMethod = fc_into->accessMethod;

	/*
	 * 创建关系。 （如果存在现有视图，则会出错，
	 * 因此我们不需要更多代码来抱怨“替换”是否为假。）
	 */
	fc_intoRelationAddr = DefineRelation(fc_create, fc_relkind, InvalidOid, NULL, NULL);

	/*
	 * 如果需要，为目标表创建一个 TOAST 表。 请注意，
	 * NewRelationCreateToastTable 以 CommandCounterIncrement() 结束，因此
	 * TOAST 表将在插入时可见。
	 */
	CommandCounterIncrement();

	/* 解析并验证 TOAST 表的 reloptions */
	fc_toast_options = transformRelOptions((Datum) 0,
										fc_create->options,
										"toast",
										fc_validnsps,
										true, false);

	(void) heap_reloptions(RELKIND_TOASTVALUE, fc_toast_options, true);

	NewRelationCreateToastTable(fc_intoRelationAddr.objectId, fc_toast_options);

	/* 创建物化视图的“视图”部分。 */
	if (fc_is_matview)
	{
		/* StoreViewQuery 在树上涂写，因此请做一个副本 */
		Query	   *fc_query = (Query *) copyObject(fc_into->viewQuery);

		StoreViewQuery(fc_intoRelationAddr.objectId, fc_query, false);
		CommandCounterIncrement();
	}

	return fc_intoRelationAddr;
}


/*
 * create_ctas_nodata
 *
 * 当使用 WITH NO DATA 时创建 CTAS 或物化视图，从
 * SELECT 的目标列表或视图定义开始。
 */
static ObjectAddress fc_create_ctas_nodata(List *fc_tlist, IntoClause *fc_into)
{
	List	   *fc_attrList;
	ListCell   *fc_t,
			   *fc_lc;

	/*
	 * 从 tlist 的非垃圾元素构建 ColumnDefs 列表。 如果在 CREATE TABLE AS
	 * 中指定了列名列表，则覆盖查询中的列名。 （列名太少是可以的，太多则不行。）
	 */
	fc_attrList = NIL;
	fc_lc = list_head(fc_into->colNames);
	foreach(fc_t, fc_tlist)
	{
		TargetEntry *fc_tle = (TargetEntry *) lfirst(fc_t);

		if (!fc_tle->resjunk)
		{
			ColumnDef  *fc_col;
			char	   *fc_colname;

			if (fc_lc)
			{
				fc_colname = strVal(lfirst(fc_lc));
				fc_lc = lnext(fc_into->colNames, fc_lc);
			}
			else
				fc_colname = fc_tle->resname;

			fc_col = makeColumnDef(fc_colname,
								exprType((Node *) fc_tle->expr),
								exprTypmod((Node *) fc_tle->expr),
								exprCollation((Node *) fc_tle->expr));

			/*
			 * 该列可能是可排序类型，但无法解析排序规则，因此请仔细检查。
			 * （我们必须在这里进行检查，因为 DefineRelation 会采用类型的
			 * 默认排序规则，而不是抱怨。）
			 */
			if (!OidIsValid(fc_col->collOid) &&
				type_is_collatable(fc_col->typeName->typeOid))
				ereport(ERROR,
						(errcode(ERRCODE_INDETERMINATE_COLLATION),
						 errmsg("no collation was derived for column \"%s\" with collatable type %s",
								fc_col->colname,
								format_type_be(fc_col->typeName->typeOid)),
						 errhint("Use the COLLATE clause to set the collation explicitly.")));

			fc_attrList = lappend(fc_attrList, fc_col);
		}
	}

	if (fc_lc != NULL)
		ereport(ERROR,
				(errcode(ERRCODE_SYNTAX_ERROR),
				 errmsg("too many column names were specified")));

	/* 使用 ColumnDef 列表创建关系定义 */
	return fc_create_ctas_internal(fc_attrList, fc_into);
}


/*
 * ExecCreateTableAs -- 执行一个 CREATE TABLE AS 命令
 */
ObjectAddress ExecCreateTableAs(ParseState *fc_pstate, CreateTableAsStmt *fc_stmt,
				  ParamListInfo fc_params, QueryEnvironment *fc_queryEnv,
				  QueryCompletion *fc_qc)
{
	Query	   *fc_query = castNode(Query, fc_stmt->query);
	IntoClause *fc_into = fc_stmt->into;
	bool		fc_is_matview = (fc_into->viewQuery != NULL);
	DestReceiver *fc_dest;
	Oid			fc_save_userid = InvalidOid;
	int			fc_save_sec_context = 0;
	int			fc_save_nestlevel = 0;
	ObjectAddress fc_address;
	List	   *fc_rewritten;
	PlannedStmt *fc_plan;
	QueryDesc  *fc_queryDesc;

	/* 检查关系是否存在 */
	if (CreateTableAsRelExists(fc_stmt))
		return InvalidObjectAddress;

	/*
	 * 创建元组接收对象并插入其所需的信息
	 */
	fc_dest = CreateIntoRelDestReceiver(fc_into);

	/*
	 * 包含的查询可以是 SELECT 或 EXECUTE 实用命令。
	 * 如果是后者，我们只需将其传递给 ExecuteQuery。
	 */
	if (fc_query->commandType == CMD_UTILITY &&
		IsA(fc_query->utilityStmt, ExecuteStmt))
	{
		ExecuteStmt *fc_estmt = castNode(ExecuteStmt, fc_query->utilityStmt);

		Assert(!fc_is_matview);	/* 语法中被排除 */
		ExecuteQuery(fc_pstate, fc_estmt, fc_into, fc_params, fc_dest, fc_qc);

		/* 获取 intorel_startup 为我们保存的对象地址 */
		fc_address = ((DR_intorel *) fc_dest)->reladdr;

		return fc_address;
	}
	Assert(fc_query->commandType == CMD_SELECT);

	/*
	 * 对于物化视图，锁定安全限制操作并安排使 GUC 变量更改仅对该命令有效。
	 * 这对于安全性不是必须的，但这使行为类似于
	 * 刷新物化视图。 否则，可能会创建一个无法刷新的物化视图。
	 */
	if (fc_is_matview)
	{
		GetUserIdAndSecContext(&fc_save_userid, &fc_save_sec_context);
		SetUserIdAndSecContext(fc_save_userid,
							   fc_save_sec_context | SECURITY_RESTRICTED_OPERATION);
		fc_save_nestlevel = NewGUCNestLevel();
	}

	if (fc_into->skipData)
	{
		/*
		 * 如果指定了 WITH NO DATA，则不通过重写器、
		 * 计划器和执行器。 只需使用类似于 CREATE VIEW 的代码路径定义关系。
		 * 这避免了在所有依赖项设置完成之前运行计划器而导致的转储/恢复问题。
		 */
		fc_address = fc_create_ctas_nodata(fc_query->targetList, fc_into);
	}
	else
	{
		/*
		 * 解析分析已经完成，但我们仍然需要运行规则
		 * 重写器。我们不进行 AcquireRewriteLocks：我们假设查询
		 * 要么直接来自解析器，要么由 plancache.c 获得了合适的锁。
		 */
		fc_rewritten = QueryRewrite(fc_query);

		/* SELECT 应该永远不会重写为多于或少于一个 SELECT 查询 */
		if (list_length(fc_rewritten) != 1)
			elog(ERROR, "unexpected rewrite result for %s",
				 fc_is_matview ? "CREATE MATERIALIZED VIEW" :
				 "CREATE TABLE AS SELECT");
		fc_query = linitial_node(Query, fc_rewritten);
		Assert(fc_query->commandType == CMD_SELECT);

		/* 计划查询 */
		fc_plan = pg_plan_query(fc_query, fc_pstate->p_sourcetext,
							 CURSOR_OPT_PARALLEL_OK, fc_params);

		/*
		 * 使用带有更新的命令 ID 的快照，以确保此查询看到
		 * 之前执行的任何查询的结果。 （这仅在计划器执行了一个
		 * 声称是稳定的函数并且改变了数据库内容时才重要，但无论如何我们都这样做，以便
		 * 与 EXPLAIN 代码路径保持平行。）
		 */
		PushCopiedSnapshot(GetActiveSnapshot());
		UpdateActiveSnapshotCommandId();

		/* 创建一个 QueryDesc，将输出重定向到我们的元组接收器 */
		fc_queryDesc = CreateQueryDesc(fc_plan, fc_pstate->p_sourcetext,
									GetActiveSnapshot(), InvalidSnapshot,
									fc_dest, fc_params, fc_queryEnv, 0);

		/* 调用 ExecutorStart 准备执行计划 */
		ExecutorStart(fc_queryDesc, GetIntoRelEFlags(fc_into));

		/* 运行计划直到完成 */
		ExecutorRun(fc_queryDesc, ForwardScanDirection, 0L, true);

		/* 如果我们给定了一个 qc 需要填充，则保存行数 */
		if (fc_qc)
			SetQueryCompletion(fc_qc, CMDTAG_SELECT, fc_queryDesc->estate->es_processed);

		/* 获取 intorel_startup 为我们保存的对象地址 */
		fc_address = ((DR_intorel *) fc_dest)->reladdr;

		/* 并进行清理 */
		ExecutorFinish(fc_queryDesc);
		ExecutorEnd(fc_queryDesc);

		FreeQueryDesc(fc_queryDesc);

		PopActiveSnapshot();
	}

	if (fc_is_matview)
	{
		/* 回滚任何 GUC 更改 */
		AtEOXact_GUC(false, fc_save_nestlevel);

		/* 恢复 userid 和安全上下文 */
		SetUserIdAndSecContext(fc_save_userid, fc_save_sec_context);
	}

	return fc_address;
}

/*
 * GetIntoRelEFlags --- 计算 CREATE TABLE AS 所需的执行器标志
 *
 * 这个是导出的，因为 EXPLAIN 和 PREPARE 也需要它。 （注意：那些
 * 调用者仍然需要显式处理 skipData 标志；由于它们
 * 使用不同的方法来抑制执行，因此尝试封装那部分似乎不值得。）
 */
int GetIntoRelEFlags(IntoClause *fc_intoClause)
{
	int			fc_flags = 0;

	if (fc_intoClause->skipData)
		fc_flags |= EXEC_FLAG_WITH_NO_DATA;

	return fc_flags;
}

/*
 * CreateTableAsRelExists --- 检查 CreateTableAsStmt 的关系是否存在
 *
 * 工具包装器检查此
 * CreateTableAsStmt 查询中待创建的关系是否已存在。 如果
 * 关系存在，则返回 true，否则返回 false。
 */
bool CreateTableAsRelExists(CreateTableAsStmt *fc_ctas)
{
	Oid			fc_nspid;
	Oid			fc_oldrelid;
	ObjectAddress fc_address;
	IntoClause *fc_into = fc_ctas->into;

	fc_nspid = RangeVarGetCreationNamespace(fc_into->rel);

	fc_oldrelid = get_relname_relid(fc_into->rel->relname, fc_nspid);
	if (OidIsValid(fc_oldrelid))
	{
		if (!fc_ctas->if_not_exists)
			ereport(ERROR,
					(errcode(ERRCODE_DUPLICATE_TABLE),
					 errmsg("relation \"%s\" already exists",
							fc_into->rel->relname)));

		/*
		 * 关系存在，并且已指定 IF NOT EXISTS。
		 *
		 * 如果我们在扩展脚本中，必须确保预先存在的
		 * 对象是扩展的成员，以避免安全风险。
		 */
		ObjectAddressSet(fc_address, RelationRelationId, fc_oldrelid);
		checkMembershipInCurrentExtension(&fc_address);

		/* 可以跳过 */
		ereport(NOTICE,
				(errcode(ERRCODE_DUPLICATE_TABLE),
				 errmsg("relation \"%s\" already exists, skipping",
						fc_into->rel->relname)));
		return true;
	}

	/* 关系不存在，可以创建 */
	return false;
}

/*
 * CreateIntoRelDestReceiver -- 创建合适的 DestReceiver 对象
 *
 * 如果从 CreateDestReceiver() 调用，intoClause 将为 NULL，
 * 在这种情况下，必须稍后提供它。 但是，允许
 * self->into 被立即填充对于其他调用者来说是方便的。
 */
DestReceiver * CreateIntoRelDestReceiver(IntoClause *fc_intoClause)
{
	DR_intorel *fc_self = (DR_intorel *) palloc0(sizeof(DR_intorel));

	fc_self->pub.receiveSlot = fc_intorel_receive;
	fc_self->pub.rStartup = fc_intorel_startup;
	fc_self->pub.rShutdown = fc_intorel_shutdown;
	fc_self->pub.rDestroy = fc_intorel_destroy;
	fc_self->pub.mydest = DestIntoRel;
	fc_self->into = fc_intoClause;
	/* 其他私有字段将在 intorel_startup 期间设置 */

	return (DestReceiver *) fc_self;
}

/*
 * intorel_startup --- 执行器启动
 */
static void fc_intorel_startup(DestReceiver *fc_self, int fc_operation, TupleDesc fc_typeinfo)
{
	DR_intorel *fc_myState = (DR_intorel *) fc_self;
	IntoClause *fc_into = fc_myState->into;
	bool		fc_is_matview;
	List	   *fc_attrList;
	ObjectAddress fc_intoRelationAddr;
	Relation	fc_intoRelationDesc;
	ListCell   *fc_lc;
	int			fc_attnum;

	Assert(fc_into != NULL);		/* 否则有人忘记设置它 */

	/* 该代码支持 CREATE TABLE AS 和 CREATE MATERIALIZED VIEW */
	fc_is_matview = (fc_into->viewQuery != NULL);

	/*
	 * 使用“预先准备好的”类型和整理信息构建列定义。如果在 CREATE TABLE AS 中指定了列名称列表，则覆盖从查询中派生的列名称。 （列名称太少是可以的，但太多则不行。）
	 */
	fc_attrList = NIL;
	fc_lc = list_head(fc_into->colNames);
	for (fc_attnum = 0; fc_attnum < fc_typeinfo->natts; fc_attnum++)
	{
		Form_pg_attribute fc_attribute = TupleDescAttr(fc_typeinfo, fc_attnum);
		ColumnDef  *fc_col;
		char	   *fc_colname;

		if (fc_lc)
		{
			fc_colname = strVal(lfirst(fc_lc));
			fc_lc = lnext(fc_into->colNames, fc_lc);
		}
		else
			fc_colname = NameStr(fc_attribute->attname);

		fc_col = makeColumnDef(fc_colname,
							fc_attribute->atttypid,
							fc_attribute->atttypmod,
							fc_attribute->attcollation);

		/*
		 * 列可能是可整理类型，但无法解决整理，因此请仔细检查。 （我们必须在这里检查，因为 DefineRelation 会采用类型的默认整理，而不是抱怨。）
		 */
		if (!OidIsValid(fc_col->collOid) &&
			type_is_collatable(fc_col->typeName->typeOid))
			ereport(ERROR,
					(errcode(ERRCODE_INDETERMINATE_COLLATION),
					 errmsg("no collation was derived for column \"%s\" with collatable type %s",
							fc_col->colname,
							format_type_be(fc_col->typeName->typeOid)),
					 errhint("Use the COLLATE clause to set the collation explicitly.")));

		fc_attrList = lappend(fc_attrList, fc_col);
	}

	if (fc_lc != NULL)
		ereport(ERROR,
				(errcode(ERRCODE_SYNTAX_ERROR),
				 errmsg("too many column names were specified")));

	/*
	 * 实际创建目标表
	 */
	fc_intoRelationAddr = fc_create_ctas_internal(fc_attrList, fc_into);

	/*
	 * 最后我们可以打开目标表
	 */
	fc_intoRelationDesc = table_open(fc_intoRelationAddr.objectId, AccessExclusiveLock);

	/*
	 * 确保构建的表未启用 RLS。
	 *
	 * 如果用户请求了无效的内容，check_enable_rls() 会自动报告 (ERROR)，否则会在 RLS 应该在这里启用时返回 RLS_ENABLED。我们目前实际上不支持这一点，因此如果发生这种情况，请抛出我们自己的报告 (ERROR)。
	 */
	if (check_enable_rls(fc_intoRelationAddr.objectId, InvalidOid, false) == RLS_ENABLED)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("policies not yet implemented for this command")));

	/*
	 * 如果它是一个物化视图并且我们要填充它，则暂时标记目标为已填充；否则，无需更改。
	 */
	if (fc_is_matview && !fc_into->skipData)
		SetMatViewPopulatedState(fc_intoRelationDesc, true);

	/*
	 * 填充 myState 的私有字段，以供后续例程使用
	 */
	fc_myState->rel = fc_intoRelationDesc;
	fc_myState->reladdr = fc_intoRelationAddr;
	fc_myState->output_cid = GetCurrentCommandId(true);
	fc_myState->ti_options = TABLE_INSERT_SKIP_FSM;

	/*
	 * 如果指定了 WITH NO DATA，则无需为批量插入设置状态，因为没有元组可插入。
	 */
	if (!fc_into->skipData)
		fc_myState->bistate = GetBulkInsertState();
	else
		fc_myState->bistate = NULL;

	/*
	 * 有效的 smgr_targblock 意味着已经有写入操作发生在该关系上。这可能是无害的，但此函数并没有为此做计划。
	 */
	Assert(RelationGetTargetBlock(fc_intoRelationDesc) == InvalidBlockNumber);
}

/*
 * intorel_receive --- 接收一个元组
 */
static bool fc_intorel_receive(TupleTableSlot *fc_slot, DestReceiver *fc_self)
{
	DR_intorel *fc_myState = (DR_intorel *) fc_self;

	/* 如果指定了 WITH NO DATA，则没有内容可插入。 */
	if (!fc_myState->into->skipData)
	{
		/*
		 * 请注意，输入槽可能不是目标关系的类型。这是由 table_tuple_insert() 支持的，但效率略低于使用正确槽插入 - 但替代方案是复制到正确类型的槽，这也不会便宜。这也不允许访问每个 AM 数据（例如元组的 xmin），但由于我们在这里不这样做...
		 */
		table_tuple_insert(fc_myState->rel,
						   fc_slot,
						   fc_myState->output_cid,
						   fc_myState->ti_options,
						   fc_myState->bistate);
	}

	/* 我们知道这是一个新创建的关系，因此没有索引 */

	return true;
}

/*
 * intorel_shutdown --- 执行者结束
 */
static void fc_intorel_shutdown(DestReceiver *fc_self)
{
	DR_intorel *fc_myState = (DR_intorel *) fc_self;
	IntoClause *fc_into = fc_myState->into;

	if (!fc_into->skipData)
	{
		FreeBulkInsertState(fc_myState->bistate);
		table_finish_bulk_insert(fc_myState->rel, fc_myState->ti_options);
	}

	/* 关闭关系，但在提交之前保持锁定 */
	table_close(fc_myState->rel, NoLock);
	fc_myState->rel = NULL;
}

/*
 * intorel_destroy --- 释放 DestReceiver 对象
 */
static void fc_intorel_destroy(DestReceiver *fc_self)
{
	pfree(fc_self);
}
